package com.atguigu.flink.sql.query;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by 黄凯 on 2023/6/26 0026 19:10
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * Flink SQL的窗口:
 * *    1. 分组窗口  groupBy
 * *         滚动窗口( 时间 、计数)
 * *         滑动窗口( 时间 、计数)
 * *         会话窗口( 时间)
 * *
 * *         定义窗口语法:
 * *            .window(Tumble.over(lit(10).minutes()).on($("rowtime")).as("w"));
 * *         使用窗口语法:
 * *            Table table = input
 * *              .window([GroupWindow w].as("w"))  // 定义窗口并指定别名为 w
 * *              .groupBy($("w"))  // 以窗口 w 对表进行分组
 * *              .select($("b").sum());  // 聚合
 * *
 * *
 * *    2. Window TVF(窗口表值函数)
 * *         滚动窗口
 * *         滑动窗口
 * *         累积窗口
 * *         会话窗口(暂不支持)
 * *
 * *    3. Over(开窗)
 */
public class Flink02_GroupWindowTableApi {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Long.valueOf(fields[1].trim()), Integer.valueOf(fields[2].trim()));
                        }

                );
        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //流转表
        Schema schema = Schema.newBuilder()
                .column("id", "STRING")
                .column("vc", "INT")
                .column("ts", "BIGINT")
                .columnByExpression("pt", "PROCTIME()")
                .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                .watermark("et", "et - INTERVAL '0' SECOND")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();

        //分组窗口
        //滚动
        //计数滚动窗口
        TumbleWithSizeOnTimeWithAlias w1 = Tumble.over(rowInterval(5L)).on($("pt")).as("w");
        //处理时间滚动窗口
        TumbleWithSizeOnTimeWithAlias w2 = Tumble.over(lit(5).seconds()).on($("pt")).as("w");
        //事件时间滚动窗口
        TumbleWithSizeOnTimeWithAlias w3 = Tumble.over(lit(5).seconds()).on($("et")).as("w");

        //滑动
        //计数滑动窗口
        //计数滑动窗口的首次计算必须要达到窗口大小。
        SlideWithSizeAndSlideOnTimeWithAlias w4 =
                Slide.over(rowInterval(5L)).every(rowInterval(3L)).on($("pt")).as("w");
        //处理时间滑动窗口
        SlideWithSizeAndSlideOnTimeWithAlias w5 =
                Slide.over(lit(10).seconds()).every(lit(5).seconds()).on($("pt")).as("w");
        //事件时间滑动窗口
        SlideWithSizeAndSlideOnTimeWithAlias w6 =
                Slide.over(lit(10).seconds()).every(lit(5).seconds()).on($("et")).as("w");

        //会话
        //处理时间会话窗口
        SessionWithGapOnTimeWithAlias w7 = Session.withGap(lit(10).seconds()).on($("pt")).as("w");
        //事件时间会话窗口
        SessionWithGapOnTimeWithAlias w8 = Session.withGap(lit(10).seconds()).on($("et")).as("w");

        //使用窗口
        table.window(w8)
                .groupBy($("w"), $("id"))
                .select($("w").start(), $("w").end(), $("id"), $("id").count())
                .execute()
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
